#include "threadpool.h"

//********************************************************************************
//初始化线程池
//********************************************************************************
void threadpoolInit(thread_pool_t *pthreadpool, int threadNum, int epfd)
{
	pthreadpool->threadNumber = threadNum;
	pthreadpool->threads = (pthread_t *)calloc(threadNum, sizeof(pthread_t));
	queueInit(&pthreadpool->queue);
	InitFileque(&pthreadpool->fileque);
	pthreadpool->userPeerfdHash = create_hashmap_int();
	pthreadpool->epfd = epfd;
	pthreadpool->Md5hashmap = create_hashmap();
}

//********************************************************************************
//销毁线程池
//********************************************************************************
void threadpoolDestroy(thread_pool_t *pthreadpool)
{
	free(pthreadpool->threads);
	queueDestroy(&pthreadpool->queue);
	DestroyFileque(&pthreadpool->fileque);
	destroy_hashmap_int(pthreadpool->userPeerfdHash);
	destroy_hashmap(pthreadpool->Md5hashmap);
}

//********************************************************************************
//创建子线程，开启线程池
//********************************************************************************
void threadpoolStart(thread_pool_t *pthreadpool)
{
	for (int idx = 0; idx < pthreadpool->threadNumber; ++idx)
	{
		int ret = pthread_create(&pthreadpool->threads[idx],
								 NULL,
								 threadFunc,
								 pthreadpool);
		THREAD_ERROR_CHECK(ret, "pthread_create");
	}
}

//********************************************************************************
//线程处理函数
//********************************************************************************
void *threadFunc(void *arg)
{
	printf("sub thread %ld is runing\n", pthread_self());
	thread_pool_t *pthreadpool = (thread_pool_t *)arg;
	while (1)
	{
		task_t *pNode = taskDequeue(&pthreadpool->queue);
		printf("我得到了任务节点,pNode->type = %d,peerid = %d\n",pNode->type,pNode->peerfd);
		if (pNode != NULL)
		{
			switch (pNode->type)
			{
			case CMD_TYPE_PUT:
				pthread_mutex_lock(&pthreadpool->fileque.mutex);
				GetFile(pNode->peerfd, pNode->data, pthreadpool);
				pthread_mutex_unlock(&pthreadpool->fileque.mutex);
				break;
			case CMD_TYPE_GET:
				PushFile(pNode->peerfd, pNode->data);
				break;
			default:
				break;
			}
		}
		else
		{
			break;
		}
		//执行完毕释放结点
		free(pNode);
	}
	printf("sub thread %ld is ready to exit\n", pthread_self());
	pthread_exit(0);
}
